package com.example.kafka.server;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.kafka.dsl.Kafka;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.ConsumerProperties;
import org.springframework.messaging.MessageChannel;

@Configuration
@EnableIntegration
public class KafkaAdapterConfig {

  private final ConsumerFactory<String, String> factory;

  public KafkaAdapterConfig(ConsumerFactory<String, String> factory) {
    this.factory = factory;
  }

  @Bean
  public IntegrationFlow kafkaIn() {
    return IntegrationFlows.from(
            Kafka.inboundChannelAdapter(factory, new ConsumerProperties("sad")),
            x -> x.poller(y -> y.fixedRate(1000)))
        .channel(kafkaIn2())
        .get();
  }

  @ServiceActivator(inputChannel = "kafkaIn2")
  void printf(String message) {
    System.out.println("message=" + message);
  }

  @Bean("kafkaIn2")
  MessageChannel kafkaIn2() {
    return new DirectChannel();
  }
}
